package org.zn.quality.handler;

import static org.zn.etl.job.context.GlobalContextKey.END_OFFSET;
import static org.zn.etl.job.context.GlobalContextKey.MAXROWS;
import static org.zn.etl.job.context.GlobalContextKey.MAX_OFFSET;
import static org.zn.etl.job.context.GlobalContextKey.PAGED_INDEX;
import static org.zn.etl.job.context.GlobalContextKey.PAGED_MODE;
import static org.zn.etl.job.context.GlobalContextKey.SRC_PK;
import static org.zn.etl.job.context.GlobalContextKey.START_OFFSET;
import static org.zn.etl.job.context.GlobalContextKey.TOTALROWS;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.EvaluationContext;
import org.springframework.stereotype.Service;
import org.zn.datax.common.entity.ColumnInfo;
import org.zn.datax.common.entity.TableInfo;
import org.zn.etl.jdbc.QueryHelper;
import org.zn.etl.jdbc.SqlBuilder;
import org.zn.etl.job.context.GlobalContextKey;
import org.zn.etl.job.entity.DataMapping;
import org.zn.etl.job.entity.EtlJob;
import org.zn.etl.job.transformer.DataTransformer;
import org.zn.etl.misc.ExpressionHelper;
import org.zn.quality.entity.FaultsData;
import org.zn.quality.entity.QualityStrategyEntity;
import org.zn.quality.entity.po.QualityResultPo;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class AllRowStrategyImpl {
	
	@Autowired
	private ExpressionHelper expressionHelper; 
	
	@Autowired
	private  Map<String, DataTransformer> dataTransformers;
	
	public  QualityResultPo	doStrategy(DataSource sSource,DataSource tSource,EtlJob job,QualityStrategyEntity stra){
		
		Map<String, Object>  context =stra.getParamMap();
		DataMapping dataMapping=job.getDataMappings().get(0);
		
		String sourceSql= dataMapping.getSourceSql();
		
		
		TableInfo srcTable= dataMapping.getSourceTable();
		TableInfo tagTable= dataMapping.getTargetTable();
		List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs  =  dataMapping.getColumnMappingPairs();
		
		String dataTrans= dataMapping.getDataTransformer();

		List<String> srcColNames=columnMappingPairs.stream().map(it-> it.getLeft().getName()).collect(Collectors.toList());
		
		if(StringUtils.isEmpty(sourceSql)) {
			sourceSql=SqlBuilder.tableSelectSql("offset","all",job.getSource().getDbType(), srcTable, srcColNames);
		}
		
		DataTransformer dtf=null;
		
		if(StringUtils.isNoneEmpty(dataTrans)) {
			dtf=dataTransformers.get(dataTrans);
		}
		
		String srcpk=srcTable.getPk().toLowerCase();
		String tagpk=tagTable.getPk().toLowerCase();
		
		context.put(SRC_PK, srcpk.toLowerCase());
	
		Long psize=(Long) Optional.ofNullable( context.get(GlobalContextKey.PAGED_SIZE)).orElse(50L);
		
		
		Long soff=(Long) Optional.ofNullable( context.get(GlobalContextKey.START_OFFSET)).orElse(50L);
		Long total=0L;
		
		Long maxOf= context.get(MAX_OFFSET)!=null? Long.valueOf(context.get(MAX_OFFSET).toString()):null ;
		Long maxRs= context.get(MAXROWS)!=null? Long.valueOf(context.get(MAXROWS).toString()):null;
		
		context.put(START_OFFSET, soff);
		
		QualityResultPo qr=new QualityResultPo();
		qr.setStartTime(new Date());
		
		List<FaultsData>  fdata=new ArrayList<>();
		
		while (true) {

			soff = soff + psize;
			context.put(END_OFFSET,soff);

				List<Map<String, Object>> sData;
				try {
					sData = loadSource(sourceSql, sSource, context, dtf, columnMappingPairs);
				if (CollectionUtils.isEmpty(sData)) {
					break;
				}
				
				String shardingKey=(String) context.get("shardingKey");
				
				List<Map<String, Object>>  tData=null;
				
				if(StringUtils.isNotEmpty(shardingKey)) {
					
					tData=new ArrayList<>();
					
					Map<Object, List<Map<String, Object>>>  keyByed=shardingKeyBy(sData, shardingKey);
					
					for (Entry<Object, List<Map<String, Object>>> entry : keyByed.entrySet()) {
						
						 tData.addAll(loadTarget(srcpk, entry.getValue(), tagTable, tSource, context, shardingKey, entry.getKey())) ;

					}	

				}else {
					tData=loadTarget(srcpk.toLowerCase(), sData, tagTable, tSource, context,null,null);
				}
				
				fdata.addAll(compareResult(sData, tData, columnMappingPairs, srcpk, tagpk));
				
				total += sData.size();
				
				if(maxOf!=null&& soff >=maxOf) {
					break;
				}
				
				if(maxRs!=null&& total >=maxRs) {
					break;
				}
				
				log.info("AllRowStrategy jobName:{}  ,run total:{} ,fdata Size :{}",job.getJobName(),total,fdata.size());
				
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
			context.put(TOTALROWS, total);   
			context.put(START_OFFSET, soff);
			
		}
		
		Map<String, Object> inspectResult =new HashMap<>();
		
		inspectResult.put("faultTotal", fdata.size());
		qr.setWarnning(fdata.size()>1?1:0); 
		qr.setInspectResult(inspectResult);
		  
		return qr;
	}
	

	
	private List<Map<String, Object>> loadSource(String expSql, DataSource sSource,Map<String, Object>  context,DataTransformer dtf,List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs) throws SQLException {
	
		String pageMode= (String) context.get(PAGED_MODE);

		EvaluationContext evalContext=expressionHelper.map2EvalContext(context);

		String  executeSql=	expressionHelper.parseExpression(expSql).getValue(evalContext, String.class);
				
		List<Map<String, Object>>  list= QueryHelper.queryMapList(sSource, executeSql);
		
		if(CollectionUtils.isEmpty(list)) {
			return null;
		}
		
		if("indexPage".equals(pageMode)) {
			
			String ipk=(String) context.get("idxPageKey");
			
			Object ipkv=list.get(list.size()-1).get(ipk.toLowerCase());
			
			if(ipkv instanceof String) {
				ipkv="'"+ipkv+"'";
			}
			
			context.put(PAGED_INDEX, ipkv);
			String frag=(String) context.get("idxPageFrag");
			context.put("idxPage", frag);
			
		}
		
		if(dtf!=null&& CollectionUtils.isNotEmpty(list)) {	
			
			dtf.transform(list, columnMappingPairs);
			
		}
		return list; 
	
	}
	
	private List<Map<String, Object>> loadTarget(String srcPk,List<Map<String, Object>> sData,TableInfo tagTable ,DataSource tSource,
			Map<String, Object>  context,String sdKey,Object sdValue) throws SQLException{
		
		StringBuilder tagSql=new StringBuilder("select * from ").append(tagTable.getSchema()).append(".").append(tagTable.getTable()).append(" where ").append(tagTable.getPk()).append(" in (");

		for (Map<String, Object> map : sData) {
			
			Object spkvalue=map.get(srcPk);
			
			if(spkvalue instanceof String) {
				
				tagSql.append("'").append(spkvalue).append("'").append(",");
				
			}
			
		}

		tagSql.delete(tagSql.length()-1, tagSql.length());
		tagSql.append(")");
		
		if(StringUtils.isNotEmpty(sdKey)) {
			
			if(sdValue instanceof String) {
				sdValue="'"+sdValue+"'";
			}
			tagSql.append("and").append(" ").append(sdKey).append("=").append(sdValue);
		}
		
		List<Map<String, Object>>  taglist= QueryHelper.queryMapList(tSource, tagSql.toString());
	
		return taglist;
		
	}
	
	private   Map<Object, List<Map<String, Object>>>  shardingKeyBy(List<Map<String, Object>> data ,String shardingKey){
		
		Map<Object, List<Map<String, Object>>>  keyByed=data.stream().collect(Collectors.groupingBy(it ->  it.get(shardingKey)));
		return keyByed;

	}
	
	private List<FaultsData> compareResult(List<Map<String, Object>> srcData,List<Map<String, Object>> tagData ,List<Pair<ColumnInfo, ColumnInfo>> cmps,String sPk,String tPd) {
	
		 Map<String, String>  columnMapping=cmps.stream().collect(Collectors.toMap( it-> it.getLeft().getName(),v->v.getRight().getName()));
		
		Map<Object, Map<String, Object >> sdataMap= srcData.stream().collect(Collectors.toMap(it-> it.get(sPk), v->v));
		
		Map<Object, Map<String, Object >> tdataMap= tagData.stream().collect(Collectors.toMap(it-> it.get(tPd), v->v));
		
		return CompareDataUtils.compareData(sdataMap, tdataMap, columnMapping);

	}
	
	
}
